package com.aotain.jupiter.abnormal;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class AbnormalReadyCheckMap implements MapFunction<Tuple2<String,AbnormalStatTuple>, Tuple2<String,AbnormalStatTuple>> {

	private HashMap<String,Long> dipPVMap ;
	private HashMap<String,Long> sdPVMap ;
	private Map<String,Long> tripleMap;
	private Map<String,Set<String>> dipToTriple;
	private Map<String,Long> delaytimeMap ;
	private Map<String,Long> maxDelayMap ;
	private Map<String,Long> minDelayMap ;
	private Map<String,Long> streamPacketMap;
	private Map<String,Long> delayTimePVMap ;
	private Map<String,Integer> abnTimeMap;
	private Map<String,Double> confidValueMap ;
	private Set<String> abnDipSet ;
	private HashMap<String,Double> dipPInfo; 
	
	@Override
	public Tuple2<String, AbnormalStatTuple> map(
			Tuple2<String, AbnormalStatTuple> value) throws Exception {
		AbnormalStatTuple stat = value.f1;
		//key = dip + "*" + idc + "*" + sip + "*" + gis + "*" + type + "*" + dport;
		String[] items = value.f0.split("\\*",-1);
		String dip = items[0];
		String idc = items[1];
		String sip = items[2];
		String gis = items[3];
		String type = items[4];
		String dport = items[5];
		long upStreamOctets = value.f1.getUpStreamOctets();
		long upStreamPacket = value.f1.getUpStreamPacket();
		String idc_dip = dip + "*" + idc ;
		String sdipKey = dip + "*" + idc + "*" + sip;
		String tripleKey = value.f0;
		
		long dipPV = 0l;
		long triplePV = upStreamOctets;
		long sdPV = 0;
		long triplePacket = upStreamPacket;
		long delaytime = value.f1.getDelayTime();
		long delayPV = 1;
		
		if(dipPVMap.get(idc_dip) != null) {
			dipPV = dipPVMap.get(idc_dip) + upStreamOctets;
		}
		if(tripleMap.get(tripleKey) != null) {
			triplePV = tripleMap.get(tripleKey) + upStreamOctets;
		}
		if(sdPVMap.get(sdipKey) != null) {
			sdPV = sdPVMap.get(sdipKey) + upStreamOctets;
		}
		if(streamPacketMap.get(tripleKey) != null) {
			triplePacket = streamPacketMap.get(tripleKey) + upStreamPacket;
		}
		if(delayTimePVMap.get(sdipKey) != null) {
			delayPV = delayTimePVMap.get(sdipKey) + 1;
		}
		if(maxDelayMap.get(sdipKey) == null) {
			maxDelayMap.put(sdipKey, delaytime);
		}
		if(delaytime>maxDelayMap.get(sdipKey)) {
			maxDelayMap.put(sdipKey, delaytime);
		}
		if(minDelayMap.get(sdipKey) == null) {
			minDelayMap.put(sdipKey, delaytime);
		}
		if(delaytime<minDelayMap.get(sdipKey)) {
			minDelayMap.put(sdipKey, delaytime);
		}
		if(delaytimeMap.get(sdipKey) != null) {
			delaytime = delaytimeMap.get(sdipKey) + delaytime;
		}

		Set<String> tripleSet = dipToTriple.get(idc_dip);
		if(tripleSet == null) {
			tripleSet = new HashSet<String>();
		}
		tripleSet.add(tripleKey);
		dipToTriple.put(idc_dip, tripleSet);

		sdPVMap.put(sdipKey, sdPV);
		dipPVMap.put(idc_dip, dipPV);
		tripleMap.put(tripleKey, triplePV);
		delaytimeMap.put(sdipKey, delaytime);
		streamPacketMap.put(tripleKey, triplePacket);
		delayTimePVMap.put(sdipKey, delayPV);
		
		
		return null;
	}

}
